#define _GNU_SOURCE
#include <errno.h>
#include <netinet/in.h>
#include <pthread.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <time.h>
#include <unistd.h>
#include <fcntl.h>
#include <uv.h>
#include "memleak/memleak.h"
#include "ringbuf/ringbuf.h"
#include "atomic/atomic.h"
#include "log.h"

#define MODE_CUSTOMER  0
#define MODE_PRODUCTOR 1
#define MODE_ECHO      2
#define MODE_PANDC     3
#define SETNONBLOCK(fd) fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK)
#define uv_close(h,c)        ({ LOGMSG("calling uv_close(%p)\n", h); uv_close(h,c); })
#define my_calloc(a,b)       ({ int _a = (int)a, _b = (int)b; void *p = calloc(_a, _b); LOGMSG("calling calloc(%u) %p\n", (int)(_a*_b), p); p; })
#define my_free(a)           ({ LOGMSG("calling free(%p)\n", a);     free(a); })
#define uv_read_start(a,b,c) ({ LOGMSG("calling uv_read_start()\n"); uv_read_start(a,b,c); })
#define uv_read_stop(a)      ({ LOGMSG("calling uv_read_stopt()\n"); uv_read_stop(a); })

#ifndef container_of
#define container_of(ptr, type, member) ({ typeof(((type *)0)->member) *__mptr = (ptr); (type *)((char *)__mptr - offsetof(type,member));})
#endif

LIST_HEAD(mem_node_list);
atomic_t ct_count = ATOMIC_INIT(0);
char g_time_str[64] ="";
int mode = MODE_ECHO;
pthread_mutex_t memleak_lock = PTHREAD_MUTEX_INITIALIZER;
struct log_handle *log_handle = NULL; int log_with_ts = 1; int log_with_func = 0, log_with_file = 0, log_with_line = 1;
uint32_t ct_count_max = 0;
uv_loop_t uv_loop;
uv_tcp_t uv_tcp_for_listen;

struct uv_tcp_buf_bundle {
    int is_read_stop;
    int uv_tcp_is_closed;
    uv_tcp_t uv_tcp;
    ringbuf_t *ringbuf;
};

typedef struct {
    uv_write_t req;
    uv_buf_t buf;
    struct uv_tcp_buf_bundle *uv_tcp_buf_bundle;
} write_req_t;
void on_write(uv_write_t *req, int status);

const char *mode2str(int mode)
{
    if (mode == MODE_PANDC    ) return "MODE_PANDC";
    if (mode == MODE_CUSTOMER ) return "MODE_CUSTOMER";
    if (mode == MODE_PRODUCTOR) return "MODE_PRODUCTOR";
    if (mode == MODE_ECHO     ) return "MODE_ECHO";
    return "unknown";
}

int mode_str2num(const char *str)
{
    if (strncmp(str, "pc", 2) == 0) return MODE_PANDC;
    if (strncmp(str, "p" , 1) == 0) return MODE_PRODUCTOR;
    if (strncmp(str, "c" , 1) == 0) return MODE_CUSTOMER;
    if (strncmp(str, "e" , 1) == 0) return MODE_ECHO;
    return -1;
}
void alloc_buffer_for_read(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    uv_tcp_t *uv_tcp = (void *)handle;
    struct uv_tcp_buf_bundle *bundle = container_of(uv_tcp, struct uv_tcp_buf_bundle, uv_tcp);
    if (mode == MODE_PANDC) {
        buf->base = (void *)bundle->ringbuf->buf;
        buf->len  = ringbuf_size(bundle->ringbuf);
    } else {
        buf->base = (void *)ringbuf_buf(bundle->ringbuf);
        buf->len  = ringbuf_get_curr_buf_len(bundle->ringbuf);
    }
}

void uv_free_cb(uv_handle_t *handle)
{
    uv_tcp_t *uv_tcp = (void *)handle;
    struct uv_tcp_buf_bundle *bundle = container_of(uv_tcp, struct uv_tcp_buf_bundle, uv_tcp);
    if (bundle) {
        if (bundle->ringbuf) destroy_ringbuf(bundle->ringbuf);
        my_free(bundle);
    }
}

void send_data(struct uv_tcp_buf_bundle *bundle)
{
    int data_len;
    void *data;
    if (mode == MODE_CUSTOMER) return;
    if (mode == MODE_PANDC) {
        data_len = ringbuf_size(bundle->ringbuf);
        data = (void *)bundle->ringbuf->buf;
    } else if (mode == MODE_PRODUCTOR) {
        data_len = ringbuf_size(bundle->ringbuf);
        data = (void *)bundle->ringbuf->buf;
    } else {
        data_len = ringbuf_get_curr_data_len(bundle->ringbuf, 0);
        data = (void *)ringbuf_data(bundle->ringbuf, 0);
    }

    if (data_len > 0) {
        write_req_t *write_req = (write_req_t*) calloc(1, sizeof(write_req_t));
        write_req->buf = uv_buf_init(data, data_len);
        write_req->uv_tcp_buf_bundle = bundle;
        if (mode == MODE_PANDC) {
        } else {
            ringbuf_pre_read_len_add(bundle->ringbuf, data_len);
        }
        uv_write(&write_req->req, (uv_stream_t *)&bundle->uv_tcp, &write_req->buf, 1, on_write);
    }
}

void on_read(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf)
{
    uv_tcp_t *uv_tcp = (void *)client;
    struct uv_tcp_buf_bundle *bundle = container_of(uv_tcp, struct uv_tcp_buf_bundle, uv_tcp);
    if (nread < 0) {
        if (nread == UV_ENOBUFS) {
            uv_read_stop(client);
            bundle->is_read_stop = 1;
        } else {
            if (bundle->uv_tcp_is_closed == 0) {
                bundle->uv_tcp_is_closed = 1;
                atomic_dec(&ct_count);
                LLOG("destory ct %p because recv %d but return %d[%s], ct_count %u[max %u]\n", bundle, (int)buf->len, (int)nread, uv_err_name(nread), atomic_read(&ct_count), ct_count_max);
                uv_close((uv_handle_t *)client, uv_free_cb);
            }
        }
        return;
    }
    if (nread > 0) {
        if (mode == MODE_ECHO) {
            ringbuf_add_data_len(bundle->ringbuf, nread);
            LOGMSG("read cache + %d ==> %d\n", (int)nread, ringbuf_get_total_data_len(bundle->ringbuf, 1));
            send_data(bundle);
        }
    }
}

void on_write(uv_write_t *req, int status)
{
    write_req_t *write_req = container_of(req, write_req_t, req);
    struct uv_tcp_buf_bundle *bundle = write_req->uv_tcp_buf_bundle;
    if (status == 0) {
        if (mode == MODE_PANDC) {
        } else if (mode == MODE_PRODUCTOR) {
        } else {
            ringbuf_pre_read_len_sub(bundle->ringbuf, write_req->buf.len);
            ringbuf_custom(bundle->ringbuf          , write_req->buf.len);
        }
        LOGMSG("read cache - %d ==> %d\n", (int)write_req->buf.len, ringbuf_get_total_data_len(bundle->ringbuf, 1));
        send_data(bundle);
        if (mode != MODE_PRODUCTOR) {
            if (bundle->is_read_stop > 0) {
                bundle->is_read_stop = 0;
                uv_read_start((uv_stream_t*) &bundle->uv_tcp, alloc_buffer_for_read, on_read);
            }
        }
    } else {
        if (bundle->uv_tcp_is_closed == 0) {
            bundle->uv_tcp_is_closed = 1;
            atomic_dec(&ct_count);
            LLOG("destory ct %p because write return %s, ct_count %u[max %u]\n", bundle, uv_err_name(status), atomic_read(&ct_count), ct_count_max);
            uv_close((uv_handle_t *)&bundle->uv_tcp, uv_free_cb);
        }
    }
    free(write_req);
}

void on_new_connection(uv_stream_t *server, int status)
{
    if (status < 0) { LOGERR("New connection error %s\n", uv_strerror(status)); exit(1); return; }
    struct uv_tcp_buf_bundle *bundle = my_calloc(1, sizeof(*bundle)); uv_tcp_init(&uv_loop, &bundle->uv_tcp);
    bundle->ringbuf = create_ringbuf(10000, 0);
    bundle->is_read_stop = 1;

    if (uv_accept(server, (uv_stream_t*) &bundle->uv_tcp)) {
        if (bundle->ringbuf) destroy_ringbuf(bundle->ringbuf);
        LOGERR("uv_accept  error\n"); my_free(bundle);
        return;
    }
    atomic_inc(&ct_count);
    uint32_t n = atomic_read(&ct_count);
    if (n > ct_count_max) ct_count_max = n;
    LLOG("ct %p created, ct count %u\n", &bundle->uv_tcp, n);
    if (mode != MODE_PRODUCTOR) {
        if (bundle->is_read_stop > 0) {
            bundle->is_read_stop = 0;
            uv_read_start((uv_stream_t*) &bundle->uv_tcp, alloc_buffer_for_read, on_read);
        }
    }
    send_data(bundle);
}

void sig_handle_usr1(int sig)
{
    memleak_stat(NULL, 0);
}

int main(int argc, char **argv)
{
    signal(SIGPIPE, SIG_IGN);
    signal(SIGUSR1, sig_handle_usr1);
    log_init(0, 0, "/tmp/blackhole.log", "/tmp/blackhole.log.2", "/tmp/blackhole.log.3");

    unsigned short port = 9000;
    struct sockaddr_in addr;
    int i;
    int ret;
    for (i = 1; i < argc; i++) {
        if (0) {
        } else if (strcmp(argv[i], "-p") == 0)     { if (i+1 >= argc) { LOGDIE("-p need arg!!\n"); } port = atoi(argv[i+1]); i++;
        } else if (strcmp(argv[i], "--mode") == 0) { if (i+1 >= argc) { LOGDIE("--mode need arg!!\n"); } mode = mode_str2num(argv[i+1]); if (mode == -1) LOGDIE("unknow mode"); i++;
        } else                                     { printf("unknown arg %s\n", argv[i]); exit(1);
        }
    }
    printf("mode %s port %u\n", mode2str(mode), port);

    uv_loop_init(&uv_loop);
    uv_tcp_init(&uv_loop, &uv_tcp_for_listen);
    uv_ip4_addr("0.0.0.0", port, &addr);
    ret = uv_tcp_bind(&uv_tcp_for_listen, (const struct sockaddr *)&addr, 0); if (ret) { LOGERR("bind error[%s]\n", uv_err_name(ret)); return -1; }
    ret = uv_listen((uv_stream_t *)&uv_tcp_for_listen, 100, on_new_connection); if (ret) { LOGERR("Listen error %s\n", uv_strerror(ret)); return ret; }
    return uv_run(&uv_loop, UV_RUN_DEFAULT);
}
